//package com.cqw.es.service;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.cqw.es.domain.EsEntity;
//import com.cqw.es.mapper.EsMapper;
//import com.cqw.order.domain.Spu;
//import com.cqw.order.mapper.SpuMapper;
//import com.rabbitmq.client.Channel;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.collections.CollectionUtils;
//import org.springframework.amqp.core.Message;
//import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
//import org.springframework.stereotype.Service;
//
//import javax.annotation.PostConstruct;
//import javax.annotation.Resource;
//import java.io.IOException;
//import java.math.BigDecimal;
//import java.util.ArrayList;
//import java.util.Date;
//import java.util.List;
//import java.util.stream.Collectors;
//
///**
// * @Author: caoqingwen
// * @CreateTime: 2021-07-20 15:12
// * @Description: 导入es
// */
//@Service
//@Slf4j
//public class EsImportService {
//
//    @Resource
//    private EsMapper esMapper;
//
//    @Resource
//    private SpuMapper spuMapper;
//
//    /**
//     * 记录导入时间
//     */
//    private Date lastImportTime;
//
//    @Resource
//    private ElasticsearchRestTemplate elasticsearchRestTemplate;
//
//
//    @PostConstruct
//    public void init() {
//        importAll();
//    }
//
//    /**
//     * 全量导入
//     */
//    public void importAll() {
//        log.info("全量导入开始=======================================");
//        List<Spu> spuList = spuMapper.selectBySelective(null);
//        if (CollectionUtils.isEmpty(spuList)) {
//            log.info("无数据需要导入");
//            return;
//        }
//
//        List<EsEntity> collect = spuList.stream().map(spu -> {
//            EsEntity esEntity = new EsEntity();
//            esEntity.setId(spu.getId());
//            esEntity.setGoodsName(spu.getSpuName());
//            esEntity.setPrice(new BigDecimal(spu.getShopId()));
//            return esEntity;
//        }).collect(Collectors.toList());
//        esMapper.saveAll(collect);
//        log.info("全量导入结束=======================================");
//
//        //记录导入时间
//        lastImportTime = new Date();
//    }
//
//
//    /**
//     * 增量导入
//     */
//    public void importUpdate() {
//        log.info("增量导入开始=============================");
//        Spu search = new Spu();
//        search.setCreateTime(lastImportTime);
//        List<Spu> spuList = spuMapper.selectBySelective(search);
//
//        if (CollectionUtils.isEmpty(spuList)) {
//            log.info("增量导入，无数据导入");
//            //更新记录时间
//            lastImportTime = new Date();
//            return;
//        }
//
//        List<EsEntity> entityList = spuList.stream().map(spu -> {
//            EsEntity esEntity = new EsEntity();
//            esEntity.setId(spu.getId());
//            esEntity.setGoodsName(spu.getSpuName());
//            esEntity.setPrice(new BigDecimal(spu.getShopId()));
//            return esEntity;
//        }).collect(Collectors.toList());
//        esMapper.saveAll(entityList);
//        //更新导入时间
//        lastImportTime = new Date();
//        log.info("增量导入结束=============================");
//    }
//
//
//    /**
//     * 快速导入
//     */
//    public void quitImport(Message message, Channel channel) {
//        //获取消息体
//        String msg = new String(message.getBody());
//        //把数据转成map集合
//        JSONObject spuInfo = JSON.parseObject(msg);
//
//        List<Long> spuIds = new ArrayList<>();
//        spuInfo.forEach((spuId, count) -> {
//            spuIds.add(Long.valueOf(spuId));
//        });
//        //查询es中的数据
//        Iterable<EsEntity> spuChangeList = esMapper.findAllById(spuIds);
//        //循环，修改数据
//        spuChangeList.forEach(esEntity -> {
//            Long id = esEntity.getId();
//
//            Integer count = spuInfo.getInteger(String.valueOf(id));
//            esEntity.setCount(esEntity.getCount() + count);
//        });
//        esMapper.saveAll(spuChangeList);
//        //签收消息
//        long deliveryTag = message.getMessageProperties().getDeliveryTag();
//        try {
//            channel.basicAck(deliveryTag, false);
//        } catch (IOException e) {
//            e.printStackTrace();
//        }
//    }
//
//}
